package com.flinksql.test;

import com.flinksql.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

import static org.apache.flink.table.api.Expressions.$;


/**
 * @author: Lin
 * @create: 2021-06-16 10:21
 * @description: 声明时间语义，基于处理时间，Table中写法
 **/
public class FlinkTableAPI_Test9_ProTime {
    public static void main(String[] args) throws Exception {
        //1.建立环境，测试不设置CK
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2.获取数据
        SingleOutputStreamOperator<WaterSensor> mapDS = env.readTextFile("input/sensor.txt")
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                });

        //3.设置处理时间字段
        Table table = tableEnv.fromDataStream(mapDS,$("id"),$("ts"),$("vc")
                                                    ,$("pt").proctime());

        table.printSchema();

    }
}
